package edu.gatech.ors.mrsim.applicationtest;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import edu.gatech.ors.mrsim.applicationtest.Assignment2.RequestFile;
import edu.gatech.ors.mrsim.basictasks.BusyTask;
import edu.gatech.ors.mrsim.basictasks.ReceiveMessage;
import edu.gatech.ors.mrsim.basictasks.SendMessage;
import edu.gatech.ors.mrsim.core.ModuleConfig;
import edu.gatech.ors.mrsim.core.ModuleSimulator;
import edu.gatech.ors.mrsim.core.SimTask;
import edu.gatech.ors.mrsim.io.conf.Configurable;
import edu.gatech.ors.mrsim.io.conf.Configuration;

public class AS2HDFS implements ModuleConfig, Configurable{

    private Configuration config;
    
    public AS2HDFS(Configuration initConfig){
	this.config = initConfig;
    }
    
    @Override
    public List<ModuleSimulator> getModuleSimulators() {
	
	//read configuration values when Andrey gets the Configuration class thing sorted out
	/*
	int numNodes = config.getInt("numNodes");
	int numReplication = config.getInt("numReplication");
	int blockSize = config.getInt("blockSize");
	double bandwidth = config.getDouble("bandwidth");
	int maxNumHandler = config.getInt("maxNumHandler");
	*/
	
	//for now just use hardcoded node number
	int numNodes = 7;
	
	//parameters for some initial files on the system
	int numFiles = 30;
	int minSize = 16;
	int maxSize = 64;
	

	List<ModuleSimulator> rv = new ArrayList<ModuleSimulator>();
	
	//add the request agent
	rv.add(new RequestAgent("requestAgent"));
	
	//cache to hold locations of the files
	Map<HDFSFile,String> fileLocations = new HashMap<HDFSFile,String>();
	
	//add each node
	for (int i = 0;i<numNodes;i++){
	    
	    String nodeName = "node"+i;
	    
	    //for testing add one file to each node
	    int fileSize = 16 + i*(64-16)/numNodes;
	    HDFSFile file = new HDFSFile("file"+(i-2)%numNodes,fileSize);
	    List<HDFSFile> files = new ArrayList<HDFSFile>();
	    files.add(file);
	    
	    //add to the file location cache
	    //that this particular file is on ths particular node
	    fileLocations.put(file,nodeName);

	    
	    rv.add(new HDFSNode(nodeName,files));
	}
	
	rv.add(new HDFSStorageManager("storageManager",fileLocations));
	
	System.out.println("Initial file locations: "+fileLocations);
	System.out.println("System setup: "+rv);
	
	// TODO Auto-generated method stub
	return rv;
    }
    
    /**
     * Abstract representation of a file.
     * @author tdoneal
     *
     */
    public static class HDFSFile {
	public int size;
	public String fileName;
	public HDFSFile(String fileName,int size){
	    this.size = size;
	    this.fileName = fileName;
	}
	
	public String toString(){
	    return "name='"+fileName+"', size="+size;
	}
    }
    
    public class HDFSStorageManager extends ModuleSimulator {

	public Map<HDFSFile,String> fileLocationCache;
	
	
	public HDFSStorageManager(String name,Map<HDFSFile,String> fileLocationCache) {
	    super(name);
	    
	    this.fileLocationCache = fileLocationCache;
	    
	    // TODO Auto-generated constructor stub
	}
	
	public String getLocationOfFile(String fileName){
	    for (HDFSFile file:fileLocationCache.keySet()){
		if (file.fileName.equals(fileName)){
		    return fileLocationCache.get(file);
		}
	    }
	    return null;
	}
	
	public void handleTaskCompletion(SimTask t){
	    super.handleTaskCompletion(t);
	    System.out.println("HDFS manager node: finished: "+t);
	    if (t instanceof ReceiveMessage){
		ReceiveMessage rm = (ReceiveMessage)t;
		Object payload = rm.packet.payload;
		
		//service file requests
		if (payload instanceof RequestFile){
		    RequestFile rf = (RequestFile)payload;
		    
		    
		    System.out.println("Searching cache for '"+rf.fileName+"'");
		    System.out.println("Cache is: "+fileLocationCache);
		    //look for file in cache
		    //simulate working here with a random pause
		    enqueueTask( new BusyTask((long)(Math.random()*5+1)));
		    if (getLocationOfFile(rf.fileName)!=null){
			//file found in cache
			//send location of the file
			this.enqueueTask(new SendMessage(getName(),rm.getSender(),new FileLocation(rf.fileName,getLocationOfFile(rf.fileName))));
		    }else{
			//send error message that file didn't exist
			this.enqueueTask(new SendMessage(getName(),rm.getSender(),"error: file '"+rf.fileName+"' does not exist in system"));
		    }
		}
	    }
	}
	
    }
    
    public static class FileLocation {
	public String fileName;
	public String location;
	public FileLocation(String fileName,String location){
	    this.fileName = fileName;
	    this.location = location;
	}
	public String toString(){
	    return "FileLocation('"+fileName+"' on "+location+")";
	}
    }
    
    public class RequestAgent extends ModuleSimulator {

	public RequestAgent(String name) {
	    super(name);
	    // TODO Auto-generated constructor stub
	}
	
	public void handleTaskCompletion(SimTask t){
	    super.handleTaskCompletion(t);
	    System.out.println("Request agent: finished: "+t);
	    if (t instanceof RequestFile){
		    this.enqueueTask(new SendMessage(getName(),"storageManager",t));
	    }
	    if (t instanceof ReceiveMessage){
		ReceiveMessage rm = (ReceiveMessage)t;
		Object payload = rm.getPayload();
		if (payload instanceof FileLocation){
		    FileLocation fl = (FileLocation)payload;
		    //send request to the appropriate node for the file
		    enqueueTask(new SendMessage(getName(),fl.location,new RequestFile(fl.fileName)));
		}
	    }
	}
	
	
    }
    
    public class HDFSNode extends ModuleSimulator {

	private List<HDFSFile> disk;
	
	public HDFSNode(String name, List<HDFSFile> filesOnDisk) {
	    super(name);
	    this.disk = filesOnDisk;
	    
	}
	
	public void handleTaskCompletion(SimTask t){
	    super.handleTaskCompletion(t);
	    System.out.println("HDFS storage node: finished: "+t);
	    if (t instanceof ReceiveMessage){
		ReceiveMessage rm = (ReceiveMessage)t;
		Object payload = ((ReceiveMessage) t).getPayload();
		if (payload instanceof RequestFile){
		    RequestFile rf = (RequestFile)payload;
		    String fileName = rf.fileName;
		    //ensure file is on disk
		    boolean fileOnDisk = false;
		    for (int i = 0;i<disk.size();i++){
			if (disk.get(i).fileName.equals(fileName)){
			    fileOnDisk = true;
			}
		    }
		    if (fileOnDisk){
			//send success message after a simulated pause
			enqueueTask(new BusyTask(20));
			enqueueTask(new SendMessage(getName(),rm.getSender(),"success: file '"+fileName+"' served to requester"));

		    }else{
			//send error message that file didn't exist
			enqueueTask(new SendMessage(getName(),rm.getSender(),"error: file '"+fileName+"' not found on local storage"));
		    }
		}
	    }
	}
	
	public String toString(){
	    return "HDFSNode("+getName()+", disk = "+disk+")";
	}
	
	
	
    }

    @Override
    public Configuration getConfiguration() {
	return config;
    }

    @Override
    public void setConfiguration(Configuration setTo) {
	this.config = setTo;
    }

}
